package com.dw.monitor.cm;



import com.dw.monitor.tool.Utc;
import com.dw.monitor.tool.jdbc;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

import static com.dw.monitor.tool.Utc.getNow;

public class YarnJsonToMysql {

    //通过http rest获取数据
    public static BufferedReader getMethod(String url) throws IOException {
        URL restURL = new URL(url);

        HttpURLConnection conn = (HttpURLConnection) restURL.openConnection();

        conn.setRequestMethod("GET"); // POST GET PUT DELETE
        conn.setRequestProperty("Accept", "application/json");

        BufferedReader br = new BufferedReader(new InputStreamReader(conn.getInputStream()));
        return br;
    }

    //解析apps数据到mysql
    public static void getAppJsonData(BufferedReader br,Connection conn) throws IOException, SQLException {
        String line;
        while((line = br.readLine()) != null ){
            JsonObject json = (JsonObject)new JsonParser().parse(line);
            //JsonArray data = json.getAsJsonArray("app");
            JsonElement value = json.get("apps").getAsJsonObject().get("app");
            JsonArray jsonElements = value.getAsJsonArray();
            if(jsonElements.size()>0){
                int sum = 0;
                for (int i = 0;i< jsonElements.size();i++){
                    JsonObject data = (JsonObject) jsonElements.get(i);
                    String id = data.get("id").toString().replace("\"","");
                    String user = data.get("user").toString().replace("\"","");
                    String queue = data.get("queue").toString().replace("\"","");
                    String state = data.get("state").toString().replace("\"","");
                    String finalStatus = data.get("finalStatus").toString().replace("\"","");
                    String applicationType = data.get("applicationType").toString().replace("\"","");
                    String startedTime = Utc.timeStampToDate(data.get("startedTime").toString());
                    String date_f = startedTime.substring(0,10);
                    String finishedTime = Utc.timeStampToDate(data.get("finishedTime").toString());
                    String elapsedTime = (data.get("elapsedTime").getAsInt()/1000+"s");
                    int runningContainers = data.get("runningContainers").getAsInt();
                    int allocatedVCores = data.get("allocatedVCores").getAsInt();
                    if(startedTime.substring(0,10).equals(getNow())) {
                        PreparedStatement psql = conn.prepareStatement("replace into dashboard_monitor_yarn_apps_data " +
                                "(id,user,queue,state,finalStatus,applicationType,startedTime,date_f,finishedTime,elapsedTime,runningContainers,allocatedVCores" +
                                ")" + "values(?,?,?,?,?,?,?,?,?,?,?,?)");
                        psql.setString(1, id);
                        psql.setString(2, user);
                        psql.setString(3, queue);
                        psql.setString(4, state);
                        psql.setString(5, finalStatus);
                        psql.setString(6, applicationType);
                        psql.setString(7, startedTime);
                        psql.setString(8, date_f);
                        psql.setString(9, finishedTime);
                        psql.setString(10, elapsedTime);
                        psql.setInt(11, runningContainers);
                        psql.setInt(12, allocatedVCores);
                        psql.executeUpdate();
                        psql.close();
                    }
                }
            }
        }
        br.close();
    }

    //解析metircs数据到mysql
    public static void getMetricsJsonData(BufferedReader br,Connection conn) throws IOException, SQLException {
        String line;
        while((line = br.readLine()) != null ){
            JsonObject json = (JsonObject)new JsonParser().parse(line);
            JsonObject value = (JsonObject) json.get("clusterMetrics");
            int appsSubmitted = value.get("appsSubmitted").getAsInt();
            int appsCompleted = value.get("appsCompleted").getAsInt();
            int appsPending = value.get("appsPending").getAsInt();
            int appsRunning = value.get("appsRunning").getAsInt();
            int appsFailed = value.get("appsFailed").getAsInt();
            int appsKilled = value.get("appsKilled").getAsInt();
            int reservedMB = value.get("reservedMB").getAsInt();
            int availableMB = value.get("availableMB").getAsInt();
            int allocatedMB = value.get("allocatedMB").getAsInt();
            int reservedVirtualCores = value.get("reservedVirtualCores").getAsInt();
            int availableVirtualCores = value.get("availableVirtualCores").getAsInt();
            int allocatedVirtualCores = value.get("allocatedVirtualCores").getAsInt();
            int containersAllocated = value.get("containersAllocated").getAsInt();
            int containersReserved = value.get("containersReserved").getAsInt();
            int containersPending = value.get("containersPending").getAsInt();
            int totalMB = value.get("totalMB").getAsInt();
            int totalVirtualCores = value.get("totalVirtualCores").getAsInt();
            int totalNodes = value.get("totalNodes").getAsInt();
            int lostNodes = value.get("lostNodes").getAsInt();
            int rebootedNodes = value.get("rebootedNodes").getAsInt();
            int activeNodes = value.get("activeNodes").getAsInt();
            //int shutdownNodes = value.get("shutdownNodes").getAsInt();
            PreparedStatement psql = conn.prepareStatement("replace into dashboard_monitor_yarn_metrics_data " +
                    "(appsSubmitted,appsCompleted,appsPending,appsRunning,appsFailed,appsKilled,reservedMB,availableMB,allocatedMB," +
                    "reservedVirtualCores,availableVirtualCores,allocatedVirtualCores,containersAllocated,containersReserved,containersPending," +
                    "totalMB,totalVirtualCores,totalNodes,lostNodes,rebootedNodes,activeNodes,shutdownNodes " +
                    ")" + "values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
            psql.setInt(1,appsSubmitted);
            psql.setInt(2,appsCompleted);
            psql.setInt(3,appsPending);
            psql.setInt(4,appsRunning);
            psql.setInt(5,appsFailed);
            psql.setInt(6,appsKilled);
            psql.setInt(7,reservedMB);
            psql.setInt(8,availableMB);
            psql.setInt(9,allocatedMB);
            psql.setInt(10,reservedVirtualCores);
            psql.setInt(11,availableVirtualCores);
            psql.setInt(12,allocatedVirtualCores);
            psql.setInt(13,containersAllocated);
            psql.setInt(14,containersReserved);
            psql.setInt(15,containersPending);
            psql.setInt(16,totalMB);
            psql.setInt(17,totalVirtualCores);
            psql.setInt(18,totalNodes);
            psql.setInt(19,lostNodes);
            psql.setInt(20,rebootedNodes);
            psql.setInt(21,activeNodes);
            psql.setInt(22,-1);
            psql.executeUpdate();
            psql.close();

        }
        br.close();
    }

    public static void main(String[] args) {
        Connection conn = jdbc.getConn();
        try {
            getAppJsonData(getMethod("http://10.10.15.61:8088/ws/v1/cluster/apps"),conn);
            getMetricsJsonData(getMethod("http://10.10.15.61:8088/ws/v1/cluster/metrics"),conn);
        } catch (IOException e1) {
            e1.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}
